package com.app.util.kafka;

import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import com.app.util.kafka.KafkaParam;

public class KafkaConsumWork implements Runnable
{
    
    private final AtomicBoolean closed = new AtomicBoolean(false);
    
    private final KafkaConsumer<String,String> consumer;
    
    private KafkaParam param;
    
    private ConsumerRecords<String,String> records;
    
    /**
     * @param kafka消费者客户端
     * @param 主题名称
     * @param 主题的分区下标
     * @param 参数
     * **/
    public KafkaConsumWork(KafkaConsumer<String,String> consumer,String topicName,int partition,KafkaParam param)
    {
        this.param    = param;
        this.consumer = consumer;
        consumer.assign(Arrays.asList(new TopicPartition(topicName, partition)));
    }
    
    @Override
    public void run()
    {
        for(;;)
        {
            try
            {
                while (!closed.get()) 
                {
                    records = consumer.poll(Duration.ofMillis(10000));
                    for(ConsumerRecord<String, String> record : records)
                    {
                        //调用对应逻辑
                        param.getMethod().invoke(param.getObj(), record.value());
                    }
                }

            }
            catch(Exception exe)
            {
                exe.printStackTrace();
            }
        }
    }

}